package com.duoec.dw.lineage.service.impl;

import com.duoec.dw.lineage.core.utils.AnalyserUtils;
import com.duoec.dw.lineage.core.utils.LineageMergeUtils;
import com.duoec.dw.lineage.dto.*;
import com.duoec.dw.lineage.dto.statement.ExportSqoopStatement;
import com.duoec.dw.lineage.dto.statement.ImportSqoopStatement;
import com.duoec.dw.lineage.dto.statement.SqoopStatement;
import com.duoec.dw.lineage.enums.DataDirectionEnum;
import com.duoec.dw.lineage.exceptions.LineageAnalyseException;
import com.duoec.dw.lineage.service.LineageAnalyseService;
import com.duoec.dw.lineage.service.LineageTableService;
import com.duoec.dw.lineage.service.MySqlSchemaService;
import com.duoec.dw.lineage.service.SqoopStatementAnalyser;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Optional;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author xuwenzhen
 */
@Service
public class LineageAnalyseServiceSqoopImpl implements LineageAnalyseService {
    private static final Logger logger = LoggerFactory.getLogger(LineageAnalyseServiceSqoopImpl.class);
    private static final Pattern TABLE_NAME_PATTERN = Pattern.compile("FROM\\s+(.+?)\\s+WHERE", Pattern.CASE_INSENSITIVE);
    private static final String TURN_LINE = "\n";
    public static final String DOT = ".";

    @Autowired
    private List<SqoopStatementAnalyser> sqoopStatementAnalyserList;

    @Autowired
    private MySqlSchemaService mySqlSchemaService;

    @Autowired
    private LineageTableService lineageTableService;

    @Autowired
    @Qualifier("lineageAnalyseServiceDruidImpl")
    private LineageAnalyseService lineageAnalyseService;

    /**
     * 解析sqoop脚本，生成血缘关系数据
     *
     * @param script 脚本字符串
     * @return 血缘关系数据
     */
    @Override
    public ScriptLineageDto analyseScript(String script) {
        List<SqoopStatement> sqoopStatementList = getSqoopStatementList(script);
        if (CollectionUtils.isEmpty(sqoopStatementList)) {
            return null;
        }

        ScriptLineageDto dto = new ScriptLineageDto();
        //处理statement
        List<TableLineageDto> lineages = Lists.newArrayList();
        for (SqoopStatement statement : sqoopStatementList) {
            if (statement instanceof ImportSqoopStatement) {
                SqoopCommendArgument args = ((ImportSqoopStatement) statement).getArgs();
                try {
                    TableLineageDto lineage = getTableLineage(args, true);
                    //追加分区
                    setPartitions(args, lineage);
                    LineageMergeUtils.addOrMerge(lineages, lineage);
                } catch (LineageAnalyseException e) {
                    dto.setErrorList(Lists.newArrayList(e.getMessage()));
                }
            } else if (statement instanceof ExportSqoopStatement) {
                SqoopCommendArgument args = ((ExportSqoopStatement) statement).getArgs();
                try {
                    TableLineageDto lineage = getTableLineage(args, false);
                    LineageMergeUtils.addOrMerge(lineages, lineage);
                } catch (LineageAnalyseException e) {
                    dto.setErrorList(Lists.newArrayList(e.getMessage()));
                }
            }
        }
        dto.setLineages(lineages);
        setVars(script, dto);
        dto.setWarningList(null);
        return dto;
    }

    private void setVars(String script, ScriptLineageDto dto) {
        dto.setVars(AnalyserUtils.analyseVars(script));
    }

    private void setPartitions(SqoopCommendArgument args, TableLineageDto lineage) {
        if (CollectionUtils.isEmpty(args.getPartitionKeys()) || lineage == null) {
            return;
        }

        args.getPartitionKeys().forEach(partition -> {
            TableFieldLineageDto field = lineage.getField(partition);
            if (field != null) {
                return;
            }
            field = new TableFieldLineageDto();
            field.setIndex(lineage.getFields().size());
            field.setColName(partition);
            field.setFieldType(2);
            lineage.getFields().add(field);
        });
    }

    private TableLineageDto getTableLineage(SqoopCommendArgument args, boolean importStatement) {
        String query = args.getQuery();
        if (StringUtils.isEmpty(query)) {
            //拉取hive和mysql的schema
            loadMySQLTableDto(args.getMySqlDatabase(), args.getMySqlTable());
            loadHiveTableDto(args.getHiveDatabase(), args.getHiveTable());
            query = args.toSql(importStatement);
        } else {
            query = fixDatabaseName(args);
        }
        TableLineageDto tableLineageDto;
        try {
            tableLineageDto = lineageAnalyseService.analyseSingleHiveSql(query);
        } catch (Exception e) {
            logger.error("解析sqoop {}失败！", importStatement ? "import" : "export", e);
            return null;
        }
        if (tableLineageDto == null) {
            return null;
        }
        DataDirectionEnum dataDirection = importStatement ? DataDirectionEnum.MYSQL_2_HIVE : DataDirectionEnum.HIVE_2_MYSQL;
        tableLineageDto.setDataDirection(dataDirection.getValue());

        if (importStatement) {
            //导入
            tableLineageDto.setDatabaseName(args.getHiveDatabase());
            tableLineageDto.setTableName(args.getHiveTable());
        } else {
            //导出
            tableLineageDto.setDatabaseName(args.getMySqlDatabase());
            tableLineageDto.setTableName(args.getMySqlTable());
        }

        return tableLineageDto;
    }

    private void loadHiveTableDto(String databaseName, String tableName) {
        TableCoordinateDto tableCoordinate = new TableCoordinateDto();
        tableCoordinate.setDatabaseName(databaseName);
        tableCoordinate.setTableName(tableName);
        TableInfoDto tableInfoDto = lineageTableService.getTableDto(tableCoordinate);
        if (tableInfoDto != null) {
            LineageTableService.addTableDto(tableInfoDto);
        }
    }

    private void loadMySQLTableDto(String databaseName, String tableName) {
        TableInfoDto tableInfoDto = mySqlSchemaService.getTableInfo(databaseName, tableName);
        if (tableInfoDto != null) {
            LineageTableService.addTableDto(tableInfoDto);
        }
    }

    /**
     * 主要是从sql脚本中抽取出表名，补全库名，提前准备好表结构信息（因为进入依赖解析时，会从hive表中拿表结构，而从sqoop导入的主要是mysql业务表！）
     * 补全库名，比如：select  * from a; 会补全成：select * from db_name.a;
     *
     * @param args 当前的导入语句的参数
     * @return 格式化好的语句
     */
    private String fixDatabaseName(SqoopCommendArgument args) {
        String query = args.getQuery();
        Matcher matcher = TABLE_NAME_PATTERN.matcher(query);
        if (!matcher.find()) {
            return query;
        }
        String tableName = matcher.group(1);
        int index = tableName.indexOf(DOT);
        if (index > -1) {
            //包含了.不需要再添加databaseName了
            loadMySQLTableDto(tableName.substring(0, index), tableName.substring(index + 1));
            return query;
        }
        String databaseName = args.getMySqlDatabase();
        loadMySQLTableDto(databaseName, tableName);
        int start = matcher.start(1);
        return query.substring(0, start) + databaseName + DOT + query.substring(start);
    }

    /**
     * 解析单条hive sql脚本，生成血缘关系数据
     *
     * @param hql 单条语句
     * @return 血缘关系数据
     */
    @Override
    public TableLineageDto analyseSingleHiveSql(String hql) {
        throw new UnsupportedOperationException("暂未实现LineageAnalyseServiceSqoopImpl.analyseSingleHiveSql方法");
    }

    private List<SqoopStatement> getSqoopStatementList(String script) {
        List<String> lines = Lists.newArrayList(Splitter.on(TURN_LINE).trimResults().split(script).iterator());

        List<SqoopStatement> sqoopStatementList = Lists.newArrayList();
        for (int i = 0; i < lines.size(); i++) {
            String line = lines.get(i);
            Optional<SqoopStatementAnalyser> analyserOpt = sqoopStatementAnalyserList
                    .stream()
                    .filter(analyser -> analyser.check(line))
                    .findFirst();
            if (analyserOpt.isPresent()) {
                SqoopStatementAnalyser analyser = analyserOpt.get();
                SqoopStatement statement = analyser.analyse(lines, i);
                if (statement == null) {
                    continue;
                }
                sqoopStatementList.add(statement);
                i = statement.getEnd();
            } else {
                logger.warn("未处理脚本：{}", line);
            }
        }
        return sqoopStatementList;
    }
}
